package com.kd.crawler.parser.thread;

import java.util.LinkedList;
import java.util.List;
import java.util.Properties;
import java.util.Queue;

import org.apache.log4j.Logger;

import com.kd.crawler.common.jms.ActiveMQProducer;
import com.kd.crawler.common.utils.PropertiesUtils;

/**
 * @author Manfred
 *
 */
public class MQMessageSender implements Runnable {
	private Logger logger=Logger.getLogger(MQMessageSender.class);
	Properties properties = PropertiesUtils.getProperties("parse.properties");
	private static Queue<MQMessage> mqMessageQueue = new LinkedList<MQMessage>();
	
	public void run() {
		while(true){
			MQMessage message=poll();
			logger.info("正在发送入口入信息："+message);
			if(message.getQueueType()==0){
				ActiveMQProducer.sendQueueMessage(message.getQueueName(), message.getMessage());
			}else{
				ActiveMQProducer.sendTopicMessage(message.getQueueName(), message.getMessage());
			}
		}
		
	}

	public static synchronized MQMessage poll() {
		MQMessage message = null;
		while (message == null) {
			message = mqMessageQueue.poll();
			if (message == null) {
				try {
					MQMessageSender.class.wait();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
		return message;
	}
	
	
	
	public static synchronized void offerQueueMessage(String queueName,String message) {
		MQMessage mqMessage=new MQMessage(queueName,0,message);
		mqMessageQueue.offer(mqMessage);
		MQMessageSender.class.notify();
	}
	
	public static synchronized void offerQueueMessages(String queueName,List<String> messages) {
		for(String message:messages){
			MQMessage mqMessage=new MQMessage(queueName,0,message);
			mqMessageQueue.offer(mqMessage);
		}
		MQMessageSender.class.notify();
	}
	
	public static synchronized void offerTopicMessage(String queueName,String message) {
		MQMessage mqMessage=new MQMessage(queueName,1,message);
		mqMessageQueue.offer(mqMessage);
		MQMessageSender.class.notify();
	}

}
